package com.pk.flink.basic;


import com.pk.flink.basic.function.PKFlatMapFunction;
import com.pk.flink.basic.function.PKMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * 使用Flink 实现单词计数
 * @author pkp://
 */
public class FlinkSocketWordCountApp {
    public static void main(String[] args) throws Exception {
        Configuration conf = new org.apache.flink.configuration.Configuration();

        //获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //读取文件内容
        //terminal to run :  nc -lk 9998
        DataStreamSource<String> stringDataSource = env.socketTextStream("localhost",9998);

        //拆分行为单词数组
        stringDataSource.flatMap(new PKFlatMapFunction()) //   aaa,bbb,ccc  =>  ["aaa","aaa","bbb"]
                        .map(new PKMapFunction()) // ["aaa","aaa","bbb"] => [("aaa",1),("aaa",1),("bbb",1)]
                        .keyBy(x->x.f0) // [("aaa",1),("aaa",1),("bbb",1)] => [("aaa",[1,1]),("bbb",[1])]
                        .sum(1)
                        .print();
        env.execute();

    }
}
